package com.iussoft.receiveData.ftp;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.apache.exception.ServiceException;
import com.apache.tools.DataMap;
import com.apache.tools.DateUtils;
import com.apache.tools.StrUtil;
import com.apache.tools.UUIDUtil;
//import com.iussoft.receive.kafka.KafkaSimpleSender;
import com.iussoft.receiveData.manager.ReceiveManager;
import com.iussoft.receiveData.schedul.ScheduledDynamicTask;
import com.iussoft.receiveData.schedul.SendService;
import com.iussoft.tools.AddressUtils;
import com.iussoft.tools.LoggerUtil;
import com.iussoft.tools.SpringContextLoader;
import org.apache.commons.io.IOUtils;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;

import java.io.InputStream;


/**
 * {
 * 	"className":"com.iussoft.ftp.FtpMessageTask",
 * 	"taskId":"ftp-find-msgtask",
 * 	"taskCorn":"0 30 18 18 10 ? ",
 * 	"data":{
 *     "requestCode":"",
 *     "directory":"",
 *     "topic":"receiveMessage"
 *        }
 * }
 */
public class FtpMessageTask implements Runnable {

    private String taskStr;
    private transient static boolean ftp_run_mark=true;
    //private KafkaSimpleSender kafkaSimpleSender;
    private ReceiveManager receiveManager;

    public FtpMessageTask(String taskStr){
        this.taskStr = taskStr;
       // this.kafkaSimpleSender = getKafka();
        this.receiveManager = getReceiveManager();
    }


    @Override
    public void run() {
        if(ftp_run_mark){
            ftp_run_mark = false;
            try {
                JSONObject jsonObj = JSON.parseObject(taskStr);
                JSONObject json = jsonObj.getJSONObject("data");
                String requestCode = json.getString("requestCode");
                String directory =json.getString("directory");
                String topic = StrUtil.doNull(json.getString("topic"),"receiveMessage");
                JSONObject dxpMap = checkPermissions(requestCode);
                String sourceMsg ="";
                DataMap<String> interfaceInfo = receiveManager.selectById("dxInterfaceReg",dxpMap.getString("fromServerId"));
                FTPClient ftpClient = getFtpClient(interfaceInfo);
                if(ftpClient == null){
                    LoggerUtil.warnSource("traceId = {}; requestTime = {}; interfaceCode = {}; requestIp = {}; errorMessage = 创建ftpClient失败; exeSys = receive; infoType = warning", UUIDUtil.getUUID("",32), DateUtils.Now.fmt_yyyyMMdd_HHmmss(), dxpMap.getString("fromInterfaceCode"), ftpClient.getRemoteAddress().getHostAddress());
                    return ;
                }
                try{
                    String interfaceCode = dxpMap.getString("fromInterfaceCode");
                    if(!ftpClient.changeWorkingDirectory(interfaceCode+"_back")){
                        ftpClient.makeDirectory(interfaceCode+"_back");
                    }
                    //String path = interfaceInfo.get("ftpPath")+"/";
                    ftpClient.changeWorkingDirectory("../"+directory);
                    FTPFile[] files = ftpClient.listFiles();
                    if(files == null && files.length == 0){
                        return;
                    }
                    for(int i=0;i<files.length;i++){
                        FTPFile file = files[i];
                        if(file.isDirectory() || file.getName().indexOf("_back")!=-1){
                           continue;
                        }
                        String traceId = UUIDUtil.getUUID("",32);
                        try {
                            JSONObject sendData = new JSONObject();
                            sendData.put("traceId", traceId);
                            sendData.put("sendTime",System.currentTimeMillis());
                            sendData.put("interfaceCode", interfaceCode);
                            sendData.put("infterfaceId", dxpMap.getString("fromServerId"));
                            sendData.put("infterfacePwd", dxpMap.getString("requestPwd"));
                            sendData.put("sourceFile", file.getName());
                            sendData.put("requestCode", requestCode);
                            sendData.put("requestUseIp", AddressUtils.getHostIp());
                            sendData.put("dataType","req");
                            InputStream in = ftpClient.retrieveFileStream(file.getName());
                            sourceMsg = IOUtils.toString(in,"UTF-8");
                            IOUtils.closeQuietly(in);
                            sendData.put("sourceMsg", sourceMsg.replaceAll("\n","").replaceAll("\r",""));
                            ftpClient.completePendingCommand();
                            SendService sendService = new SendService(sendData);
                            ScheduledDynamicTask.getInstance().submit(sendService);
                            //kafkaSimpleSender.sendMsg(topic, "0", sendData.toJSONString());
                            ftpClient.rename(file.getName(),"../"+interfaceCode+"_back/"+file.getName());
                            LoggerUtil.warnSource("traceId = {}; requestUrl = {}; requestTime = {}; interfaceCode = {}; requestIp = {}; requestBody = {}; exeSys = receive; infoType = info", traceId, "ftp", DateUtils.Now.fmt_yyyyMMdd_HHmmss() + "", interfaceCode, ftpClient.getRemoteAddress().getHostAddress(), sourceMsg.replaceAll("\n","").replaceAll("\r","").replaceAll(";","；"));
                        }catch (Exception e){//requestTime = {}; requestIp = {};
                            LoggerUtil.errorSource("exception = Exception; fileName = {}; traceId = {}; filePath = {}; errorMessage = {}; exeSys = receive; level = {}; infoType = exception",file.getName(),traceId,ftpClient.printWorkingDirectory(),e.getMessage(),"ERROR");
                        }
                    }
                } finally {
                    if(ftpClient != null) {
                        if (ftpClient.isConnected()) {
                            ftpClient.disconnect();
                        }
                    }
                }
            }catch (Exception e){
                LoggerUtil.error(taskStr+"; FTP报文获取异常：", e);
            } finally {
                ftp_run_mark=true;
            }
        }
    }

    private FtpManager getManager(){
        return (FtpManager) SpringContextLoader.getBean("ftpManager");
    }

    private FTPClient getFtpClient(DataMap<String> interfaceInfo){
        if(interfaceInfo==null){
            return null;
        }
        try {
            FTPClient ftpClient = new FTPClient();
            ftpClient.setConnectTimeout(2000);
            ftpClient.connect(interfaceInfo.get("interfaceUri"), Integer.valueOf(interfaceInfo.get("interfacePort")));
            boolean result = ftpClient.login(interfaceInfo.get("ftpUserName"), interfaceInfo.get("ftpPwd"));
            if (!result) {
                LoggerUtil.error("ftpClient登陆失败! userName:" + interfaceInfo.get("ftpUserName") + " ; password:" + interfaceInfo.get("ftpPwd"));
            }
            ftpClient.setBufferSize(1024);
            ftpClient.setControlEncoding("UTF-8");
            ftpClient.changeWorkingDirectory(interfaceInfo.get("ftpPath"));
            return ftpClient;
        }catch (Exception e){
            LoggerUtil.error("创建FTPClient失败",e);
        }
        return null;
    }

    private ReceiveManager getReceiveManager(){
        return (ReceiveManager) SpringContextLoader.getBean("receiveManager");
    }

    /*private KafkaSimpleSender getKafka(){
        return (KafkaSimpleSender) SpringContextLoader.getBean("kafkaSimpleSender");
    }*/

    private JSONObject checkPermissions(String interfaceCode) throws ServiceException {
        JSONObject data = receiveManager.getDxSysIntRelation(interfaceCode);
        if(null != data && !data.isEmpty()){
            return data;
        }
        return null;
    }
}
